package com.zhu.utils;


import com.zhu.bean.TransientSink;
import com.zhu.config.ClickHouseConfig;
import lombok.SneakyThrows;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * DWS 层 将数据写入到clickHouse
 */
public class ClickHouseUtil {

    //T 输入数据类型
    public static <T> SinkFunction<T> getClickHouseSinkFunction(String sql) {

        return JdbcSink.<T>sink(sql,
                new JdbcStatementBuilder<T>() {
                    @SneakyThrows
                    @Override
                    public void accept(PreparedStatement preparedStatement, T t) throws SQLException {  //预编译sql 占位符赋值
                        Class<?> tClz = t.getClass();
                        Field[] declaredFields = tClz.getDeclaredFields();  //获取所有方法
                        int offset = 0;
                        for (int i = 0; i < declaredFields.length; i++) {
                            Field field = declaredFields[i];  //获取单个方法对象  属性调用 Object Class
                            //获取单个属性
                            field.setAccessible(true);

                            TransientSink annotation = field.getAnnotation(TransientSink.class);
                            if (annotation != null){
                                offset ++;
                                continue;
                            }
                            Object value = field.get(t);   //获取单个属性值
                            //给站位符赋值
                            preparedStatement.setObject(i + 1 - offset, value);
                        }
                    }
                },
                JdbcExecutionOptions.builder()
                        .withBatchSize(5)
                        .withBatchIntervalMs(1000L)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(ClickHouseConfig.CLICKHOUSE_DRIVER)
                        .withUrl(ClickHouseConfig.CLICKHOUSE_URL)
                        .withUsername(ClickHouseConfig.CLICKHOUSE_USER)
                        .withPassword(ClickHouseConfig.CLICKHOUSE_PASSWORD)
                        .build()
                );
    }

    //如果javaBean中含有辅助字段   多于sql表中的属性字段

}
